package com.atguigu.day09;

import com.atguigu.util.UserBehavior;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;

public class Example7 {
    public static void main(String[] args) throws Exception {
        var env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 实例化一个数据源
        var source = FileSource
                .forRecordStreamFormat(
                        // 按行消费
                        new TextLineInputFormat(),
                        new Path("/home/yuantuzhi/flinktutorial0905/src/main/resources/UserBehavior.csv")
                )
                .build();

        var stream = env
                .fromSource(source, WatermarkStrategy.noWatermarks(), "userbehavior")
                .map(new MapFunction<String, UserBehavior>() {
                    @Override
                    public UserBehavior map(String in) throws Exception {
                        var fields = in.split(",");
                        return new UserBehavior(fields[1], fields[3], Long.parseLong(fields[4]) * 1000L);
                    }
                })
                .filter(r -> r.type.equals("pv"))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                                .withTimestampAssigner(new SerializableTimestampAssigner<UserBehavior>() {
                                    @Override
                                    public long extractTimestamp(UserBehavior element, long recordTimestamp) {
                                        return element.ts;
                                    }
                                })
                );

        // 获取表执行环境
        var streamTableEnvironment = StreamTableEnvironment.create(
                env,
                EnvironmentSettings.newInstance().inStreamingMode().build()
        );

        // 将数据源转换成动态表
        var dynamicTable = streamTableEnvironment
                .fromDataStream(
                        stream,
                        $("productId").as("product_id"),
                        $("type"),
                        $("ts").rowtime() // rowtime表示这一列是事件时间列
                );

        // 需要将动态表注册为临时视图
        streamTableEnvironment.createTemporaryView("user_behavior", dynamicTable);

        // 滑动窗口：HOP(时间戳, 滑动距离, 窗口长度)
        // 滚动窗口：TUMBLE(时间戳，窗口长度)
        // stream.keyBy(r -> r.productId).window(SlidingEventTimeWindow.of(....)).aggregate(new AggregateFunction, new ProcessWindowFunction)
        // COUNT是将窗口中所有数据收集然后计数
        var resultTable = streamTableEnvironment
                .sqlQuery(
                        "SELECT product_id, COUNT(product_id) AS pv_count, " +
                                " HOP_START(ts, INTERVAL '5' MINUTES, INTERVAL '1' HOURS) AS window_start_time, " +
                                " HOP_END(ts, INTERVAL '5' MINUTES, INTERVAL '1' HOURS) AS window_end_time" +
                                " FROM user_behavior GROUP BY product_id, HOP(ts, INTERVAL '5' MINUTES, INTERVAL '1' HOURS)"
                );

        streamTableEnvironment.toChangelogStream(resultTable).print();

        env.execute();
    }
}
